package desgin.pattern.other.reactor;

import java.io.IOException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * Created by kai on 2015/6/28 0028.
 */
public class NioReactor {
    private String name;//reactor name
    private R reactorR;
    private W reactorW;

    public NioReactor(String name) throws IOException {
        this.name = name;
        this.reactorR = new R();
        this.reactorW = new W();
    }

    final void startup() {
        new Thread(reactorR).start();
        new Thread(reactorW).start();
        System.out.println("reactor 读写线程已经开启...");
    }

    final BlockingQueue<NioConnection> getRegisterQueue() {
        return reactorR.registerQueue;
    }

    final BlockingQueue<NioConnection> getWriteQueue() {
        return reactorW.writeQueue;
    }

    final void postRegister(NioConnection nioConnection) {
        System.out.println("流转到reactor的registerQueue");
        reactorR.registerQueue.offer(nioConnection);
        reactorR.selector.wakeup();
        System.out.println("已唤醒读线程");
    }

    final void postWrite(NioConnection nioConnection) {
        reactorW.writeQueue.offer(nioConnection);
    }

    private final class R implements Runnable {
        private final Selector selector;
        private final BlockingQueue<NioConnection> registerQueue;
        private long reactCount;

        public R() throws IOException {
            this.selector = Selector.open();
            this.registerQueue = new LinkedBlockingDeque<>();
        }

        @Override
        public void run() {
            final Selector selector = this.selector;
            for (; ; ) {
                try {
                    System.out.println("reactor R 轮训");
                    selector.select(100000L);
                    register(selector);
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    for (SelectionKey key : selectionKeys) {
                        Object att = key.attachment();
                        System.out.println("att 已经得到............");
                        if (att != null && key.isValid()) {
                            int readyops = key.readyOps();
                            if ((readyops & SelectionKey.OP_READ) != 0) {
                                read((NioConnection) att);
                            } else if ((readyops & SelectionKey.OP_WRITE) != 0) {
                                write((NioConnection) att);
                            } else {
                                key.cancel();
                            }
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }


        private void register(Selector selector) throws ClosedChannelException {
            System.out.println("register start");
            NioConnection c = null;
            c = registerQueue.poll();
            if (c != null) {
                c.register(selector);
            }
        }

        private void read(NioConnection nioConnection) {
            System.out.println("read start ..reactor");
            nioConnection.read();
        }

        private void write(NioConnection nioConnection) {
            System.out.println("write start ...reactor");
            nioConnection.write();
        }
    }

    private final class W implements Runnable {
        private BlockingQueue<NioConnection> writeQueue;

        public W() {
            writeQueue = new LinkedBlockingDeque<>();
        }

        @Override
        public void run() {
            NioConnection c = null;
            for (; ; ) {
                try {
                    c = writeQueue.take();
                    if (c != null) {
                        c.write();
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        }
    }
}
